package com.gitee.cnlongs.cnlongmq.server.threads.engine;

import java.util.List;

import org.apache.commons.collections.CollectionUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.gitee.cnlongdb.basic.entity.database.LongTableEntity;
import com.gitee.cnlongdb.basic.entity.query.LongDbQuery;
import com.gitee.cnlongdb.basic.entity.query.LongDbWhere;
import com.gitee.cnlongdb.monitor.table.CnLongTableSummary;
import com.gitee.cnlongdb.server.connect.LongDbConnect;
import com.gitee.cnlongs.cnlongmq.basic.constant.CnLongMqConstant;
import com.gitee.cnlongs.cnlongmq.basic.entitiy.request.CnLongMqMessage;
import com.gitee.cnlongs.cnlongmq.basic.enums.CnLongMqDataSync;
import com.gitee.cnlongs.cnlongmq.basic.enums.CnLongMqStatus;
import com.gitee.cnlongs.cnlongmq.server.LongMqServer;
import com.gitee.cnlongs.cnlongmq.server.handler.LongMqRequest;

/**
 * 系统名称：AgainSendFailMqMessageThread
 * 模块名称：
 * 模块描述：开启【消息发送失败】重新发送消息的监听
 * 功能列表：
 * 模块作者：zouyong
 * 开发时间：2019年4月18日 上午11:19:24
 * 模块路径：com.gitee.cnlongs.cnlongmq.server.threads.engine.AgainSendFailMqMessageThread
 */
public class AgainSendFailMqMessageThread extends Thread {

	private static final Logger logger = (Logger) LoggerFactory.getLogger(AgainSendFailMqMessageThread.class);
	
	private LongDbConnect connect;
	
	public AgainSendFailMqMessageThread(LongDbConnect connect) {
		this.connect = connect;
	}
	
	@Override
	public void run() {
		try {
			// 获取当前数据库下所有的表信息列表
			List<LongTableEntity> tableList = CnLongTableSummary.getTableListByConnect(connect);
			if(CollectionUtils.isEmpty(tableList)) {
				return;
			}
			
			// 查询所有失败的消息
			LongDbQuery query = new LongDbQuery();
			query.addLongDbWhere(LongDbWhere.where(CnLongMqDataSync.longMqStatus.name()).is(CnLongMqStatus.failure.name()).tableMatching(true));
			
			// 遍历查询所有表发送失败的消息
			for (LongTableEntity table : tableList) {
				query.setTableName(table.getTableName());
				query.setClassz(CnLongMqMessage.class);
				long totalCount = LongMqServer.getExecuter().getLongDbServer().count(query); // 查询总数
				query.setPageSize(Integer.valueOf(totalCount+""));
				List<CnLongMqMessage> list = null;
				try {
					list = LongMqServer.getExecuter().getLongDbServer().find(query);
				} catch (Exception e) {
					logger.error("查询表[{}]发送失败的消息数据异常：{}", table.getTableName(), e.getMessage());
					continue;
				}
				if(CollectionUtils.isEmpty(list)) {
					continue;
				}
				for (CnLongMqMessage mqMessage : list) {
					long surplusTime = System.currentTimeMillis() - mqMessage.getSendDate().getTime();
					// 判断消息是否过期
					if(surplusTime >= CnLongMqConstant.millisecond * mqMessage.getLongMqTimeOut()) {
						LongMqRequest.processTransferMqMessage(mqMessage, CnLongMqStatus.discard.name());
					}
					// 判断是否到达重发时间
					else if(surplusTime >= CnLongMqConstant.millisecond * CnLongMqConstant.defaultAgainSendFailMqMessageTime) {
						try {
							mqMessage.setSendTimes(mqMessage.getSendTimes()+1);
							mqMessage.setAgainSend(true); // 消息重发
							LongMqServer.getExecuter().againSendLongMqMessage(mqMessage, false);
						} catch(Exception e) {
							logger.error("消息队列：{}，重发异常：{}", mqMessage.getLongMqKey(), e.getMessage());
						}
					}
				}
			}
		}
		catch(Exception e) {
			logger.error(e.getMessage());
		}
	}
	
}
